package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-11 14:49
 */
public class Dwd_07_TradeCancelDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_07_TradeCancelDetail().init(
                7001,
                3,
                "Dwd_07_TradeCancelDetail");
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        readOrderPreProcessFromKafka(tableEnv, "Dwd_07_TradeCancelDetail");

        Table resultTable = tableEnv.sqlQuery(
                "select  " +
                "    `id`,  " +
                "    `order_id`,  " +
                "    `sku_id`,  " +
                "    `sku_name`,  " +
                "    `create_time`,  " +
                "    `source_id`,  " +
                "    `source_type_code`,  " +
                "    `source_type_name`,  " +
                "    `sku_num`,  " +
                "    `split_origin_amount`,  " +
                "    `split_total_amount`,  " +
                "    `split_activity_amount`,  " +
                "    `split_coupon_amount`,  " +
                "    `od_ts`,  " +
                "    `user_id`,  " +
                "    `province_id`,  " +
                "    `operate_time`,  " +
                "    `order_status`,  " +
                "    `oi_ts`,  " +
                "    `activity_id`,  " +
                "    `activity_rule_id`,  " +
                "    `coupon_id`,  " +
                "    `row_opt_ts`  " +
                "from order_pre_process  " +
                "where `type` = 'update' and `old`['order_status'] is not null and `order_status` = '1003'"
        );
        
        tableEnv.executeSql(
                "create table dwd_trade_cancel_detail (  " +
                "    id string,  " +
                "    order_id string,  " +
                "    sku_id string,  " +
                "    sku_name string,  " +
                "    create_time string,  " +
                "    source_id string,  " +
                "    source_type_code string,  " +
                "    source_type_name string,  " +
                "    sku_num string,  " +
                "    split_origin_amount string,  " +
                "    split_total_amount string,  " +
                "    split_activity_amount string,  " +
                "    split_coupon_amount string,  " +
                "    od_ts bigint,  " +
                "    user_id string,  " +
                "    province_id string,  " +
                "    operate_time string,  " +
                "    order_status string,  " +
                "    oi_ts bigint,  " +
                "    activity_id string,  " +
                "    activity_rule_id string,  " +
                "    coupon_id string,  " +
                "    row_opt_ts timestamp_ltz(3)  " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_TRADE_CANCEL_DETAIL)
        );

        resultTable.executeInsert("dwd_trade_cancel_detail");
    }
}
